package com.navinfo.tripanalysis.offline.service.impl;

import com.navinfo.tripanalysis.common.arithmetic.common.DrivingData;
import com.navinfo.tripanalysis.common.util.DateUtils;
import com.navinfo.tripanalysis.common.util.JsonUtils;
import com.navinfo.tripanalysis.offline.service.DrivingDataService;
import com.navinfo.tripanalysis.offline.util.BigDataUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.catalyst.expressions.GenericRow;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import scala.Tuple2;

import java.net.URI;
import java.util.*;

/**
 * DrivingData缓存服务
 * @author 沈东生
 */
public class DrivingDataServiceImpl extends AbstractHiveSaveService implements DrivingDataService {

    @Override
    public JavaPairRDD<Long, DrivingData> load(SparkSession spark, JavaSparkContext jsc, long date) {
        //确定昨日产生的DrivingData数据的位置
        String day = DateUtils.format(new Date(date - 86400000), DateUtils.DateFormat.YYYYMMDD);

        if(BigDataUtils.SaveType.HIVE.equals(saveType)){
            StringBuilder sb = new StringBuilder("select tid, data_json");
            sb.append(" from ");
            if(StringUtils.isNotEmpty(getHiveDbName())){
                sb.append(getHiveDbName()).append(".");
            }
            sb.append(getHiveTableName());
            sb.append(" where day='"+day+"'");

            String sql = sb.toString();
            logger.error("load drivingData sql:-----------------\n{}", sql);
            return spark.sql(sql).toJavaRDD().mapPartitionsToPair((PairFlatMapFunction<Iterator<Row>, Long, DrivingData>) DrivingDataServiceImpl::getDrivingData);

        }else if(BigDataUtils.SaveType.HDFS.equals(saveType)) {
            String dataPath = hdfsPath + "/day=" + day;

            try {
                Path path = new Path(dataPath);
                FileSystem fs = FileSystem.get(new URI(dataPath), new Configuration());
                if (!fs.exists(path)) {
                    JavaRDD<Row> rdd = jsc.parallelize(new ArrayList<>());
                    return rdd.mapPartitionsToPair((PairFlatMapFunction<Iterator<Row>, Long, DrivingData>) DrivingDataServiceImpl::getDrivingData);
                }
            } catch (Exception e) {
                logger.error("load drivingData hdfs error...",e);
            }

            return spark.read().parquet(dataPath).toJavaRDD().mapPartitionsToPair((PairFlatMapFunction<Iterator<Row>, Long, DrivingData>) DrivingDataServiceImpl::getDrivingData);
        }else{
            System.out.println("没有配置保存数据到hive的保存类型");
        }

        return null;
    }

    @Override
    public void save(SparkSession spark, JavaSparkContext jsc, JavaPairRDD<Long, DrivingData> rdd, long date) {
        //保存数据
        long start = System.currentTimeMillis();
        logger.error("进行DrivingData信息处理，保存数据开始...");

        String day = DateUtils.format(new Date(date), DateUtils.DateFormat.YYYYMMDD);
        super.saveData(spark, rdd, hdfsPath +"/day="+day, false, Collections.singletonList("day"),  Collections.singletonList("'"+day+"'"));

        logger.error("进行DrivingData信息处理，保存数据耗时{}ms", System.currentTimeMillis()-start);
    }

    @Override
    protected StructType createSchema() {
        List<StructField> list = new ArrayList<>();
        list.add(new StructField("oid", DataTypes.LongType, true, Metadata.empty()));
        list.add(new StructField("tid", DataTypes.LongType, true, Metadata.empty()));
        list.add(new StructField("data_json", DataTypes.StringType, true, Metadata.empty()));
        list.add(new StructField("update_day", DataTypes.StringType, true, Metadata.empty()));
        list.add(new StructField("cross_day_flag", DataTypes.IntegerType, true, Metadata.empty()));
        return new StructType(list.toArray(new StructField[list.size()]));
    }

    @Override
    protected JavaRDD<Row> createRowRDD(JavaPairRDD resourceRDD) {
        return resourceRDD.mapPartitions((FlatMapFunction<Iterator<Tuple2<Long, DrivingData>>, Row>) tuple2It -> new Iterator<Row>() {
            @Override
            public boolean hasNext() {
                return tuple2It.hasNext();
            }

            @Override
            public Row next() {
                Tuple2<Long, DrivingData> tuple2 = tuple2It.next();
                DrivingData m = tuple2._2;

                List<Object> list = new ArrayList<>();
                list.add(tuple2._1);
                list.add(tuple2._1);
                list.add(JsonUtils.toJsonStr(m));
                list.add(m.getUpdateDay());
                list.add(m.getCrossDayFlag());

                return new GenericRow(list.toArray());
            }
        });
    }

    private static Iterator<Tuple2<Long, DrivingData>> getDrivingData(Iterator<Row> iterator) {
        return new Iterator<Tuple2<Long, DrivingData>>() {
            @Override
            public boolean hasNext() {
                return iterator.hasNext();
            }
            @Override
            public Tuple2<Long, DrivingData> next() {
                Row row = iterator.next();
                long tid = row.getLong(row.fieldIndex("tid"));
                return new Tuple2<>(tid, JsonUtils.fromJson(row.getString(row.fieldIndex("data_json")), DrivingData.class));
            }
        };
    }


}
